package com.explorex.infra.puma.gateway.netty.rrmp;

import com.explorex.infra.puma.gateway.netty.common.*;
import com.explorex.infra.puma.gateway.netty.common.enums.EnumTerminal;
import com.explorex.infra.puma.gateway.netty.common.enums.EnumUserStatus;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;

import java.nio.channels.ClosedChannelException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author: JacceYang
 * @createTime: 2022/09/03 17:03
 * @description: rrmp 协议client session 信息
 */
@Slf4j
public class RTcpSession implements BaseSocketSession, BaseSessionAttribute {


    private final RTCPServer rtcpServer;
    private final Channel channel;
    private final ChannelManager channelManager;
    private final AtomicInteger seqId = new AtomicInteger(0);
    /**
     * 鉴权状态,初始化时为未鉴权状态
     */
    private volatile AuthStatusEnum auth = AuthStatusEnum.NOT_AUTH;
    private String domain;
    private String remoteIp;

    private int remotePort;
    private String localIp;
    /**
     * user status 1- online ,2- offline ,3-background
     */
    private int status = EnumUserStatus.OFFLINE.getKey();
    private int preStatus = EnumUserStatus.OFFLINE.getKey();
    private DeviceInfo deviceInfo;
    private int terminal;
    private int channelTag;
    private byte[] shareKey;
    private String activeUserId;
    // 最近一次收到消息上行的时间
    private volatile long lastTime = System.currentTimeMillis();

    /**
     * 网络连接 四元组
     */
    private String ipLink;

    /**
     * 是否需要记录 日志
     */
    private boolean needLog;

    private UserKey authUser;
    /**
     * refresh时间周期索引
     */
    private int refreshIndex;

    @Override
    public UserKey getAuthUser() {
        return authUser;
    }

    public void setAuthUser(UserKey authUser) {
        this.authUser = authUser;
    }

    protected Map<String, UserKey> tokenUserInfo = new ConcurrentHashMap<>();

    @Override
    public Map<String, UserKey> getTokenUserInfo() {
        return tokenUserInfo;
    }

    /**
     * 鉴权的时候带过来的上下文信息
     */
    Map<String, String> extra;

    @Override
    public Map<String, String> getExtra() {
        return extra;
    }

    /**
     * 是否熔断ack 重试逻辑，对于消息没有回复的ack 重试，极端场景关闭重试导致的client端的压力增大，例如弱网或者瞬间大流量，由于重试存在导致消息压力大。
     */
    private volatile boolean ackRetryFuse;

    public boolean isAckRetryFuse() {
        return ackRetryFuse;
    }

    public void setAckRetryFuse(boolean ackRetryFuse) {
        this.ackRetryFuse = ackRetryFuse;
    }

    @Override
    public boolean isRateLimited() {
        return false;
    }

    public RTcpSession(Channel channel, ChannelManager channelManager, RTCPServer rtcpServer) {
        this.channel = channel;
        this.channelManager = channelManager;
        this.rtcpServer = rtcpServer;
    }

    private ChannelFuture sendBinaryMessage(ByteBuf byteBuf) {
        return channel.writeAndFlush(byteBuf);
    }

    @Override
    public ChannelFuture sendMessage(Object msg) {
        long startTime = System.currentTimeMillis();
        // 写之前判断一下通道是否可写
        if (!canWrite()) {
            if (!channel.isOpen()) {
                closeSocket("is close");
                return channel.newFailedFuture(new ClosedChannelException());
            }
            return channel.newSucceededFuture();
        }

        if (msg instanceof ByteBuf) {
            return sendBinaryMessage((ByteBuf) msg);
        }
        throw new UnsupportedOperationException();
    }

    @Override
    public ChannelFuture closeSocket(String msg) {
        if (canWrite()) {
            channel.flush();
        }

        return channel.close();
    }

    @Override
    public Channel channel() {
        return channel;
    }

    @Override
    public String domain() {
        return domain;
    }

    @Override
    public String remoteIp() {
        return remoteIp;
    }

    @Override
    public int remotePort() {
        return remotePort;
    }

    public void setRemotePort(int remotePort) {
        this.remotePort = remotePort;
    }

    @Override
    public String localIp() {
        return localIp;
    }

    @Override
    public String ipLink() {
        return ipLink;
    }

    @Override
    public DeviceBase device() {
        return deviceInfo;
    }

    @Override
    public String userId() {
        return activeUserId;
    }

    @Override
    public int userStatus() {
        return status;
    }

    @Override
    public int preStatus() {
        return preStatus;
    }

    @Override
    public int terminal() {
        return terminal;
    }

    @Override
    public int channelTag() {
        return channelTag;
    }

    @Override
    public boolean needLog() {
        return needLog;
    }

    public void setNeedLog(boolean needLog) {
        this.needLog = needLog;
    }

    @Override
    public ProtocolEnum protocol() {
        return ProtocolEnum.RRPM;
    }

    @Override
    public boolean canWrite() {
        return channel.isActive() && channel.isWritable();
    }

    public void setDomain(String domain) {
        this.domain = domain;
    }

    public int nextSeqId() {
        return seqId.getAndAdd(1);
    }

    public DeviceInfo getDeviceInfo() {
        return deviceInfo;
    }

    public AuthStatusEnum getAuth() {
        return auth;
    }

    @Override
    public void setAuth(AuthStatusEnum auth) {
        this.auth = auth;
    }

    public String getRemoteIp() {
        return remoteIp;
    }

    public void setRemoteIp(String remoteIp) {
        this.remoteIp = remoteIp;
    }

    public ChannelManager getChannelManager() {
        return channelManager;
    }


    @Override
    public void setStatus(int status) {
        this.preStatus = this.status;
        this.status = status;
    }

    public String getLocalIp() {
        return localIp;
    }

    public void setLocalIp(String localIp) {
        this.localIp = localIp;
    }

    public String getIpLink() {
        return ipLink;
    }

    public void setIpLink(String ipLink) {
        this.ipLink = ipLink;
    }

    public int getTerminal() {
        return terminal;
    }

    public int getChannelTag() {
        return channelTag;
    }

    public void setChannelTag(int channelTag) {
        this.channelTag = channelTag;
    }

    public void refresh(boolean force) {
        // 如果消息有上行，也可以更新心跳
        if (this.auth == AuthStatusEnum.AUTHED && (force || System.currentTimeMillis() - lastTime > 5* 1000)) {
            lastTime = System.currentTimeMillis();
            this.rtcpServer.getBaseSessionRedis().updateSessionInfo(this);
        }
    }

    @Override
    public long getLastTime() {
        return lastTime;
    }

    @Override
    public void setLastTime(long lastTime) {
        this.lastTime = lastTime;
    }

    public String getActiveUserId() {
        return activeUserId;
    }

    public void setActiveUserId(String activeUserId) {
        this.activeUserId = activeUserId;
    }

    public String token(String userId) {
        return String.format("%s@%s@%s", userId, this.domain, this.terminal);
    }

    public byte[] getShareKey() {
        return shareKey;
    }

    public void setShareKey(byte[] shareKey) {
        this.shareKey = shareKey;
    }

    @Override
    public String toString() {
        return "RTcpSession{" +
                "channelId=" + channel.id().asLongText() +
                ", auth=" + auth +
                ", domain='" + domain + '\'' +
                ", remoteIp='" + remoteIp + '\'' +
                ", status=" + status +
                ", preStatus=" + preStatus +
                ", device=" + deviceInfo +
                ", lastTime=" + lastTime +
                ", ipLink='" + ipLink + '\'' +
                ", authUser=" + authUser +
                ", tokenUserInfo=" + tokenUserInfo +
                '}';
    }
}
